package com.example.toy.core.conf;

import com.example.toy.core.partition.PartitionPathStyle;
import com.example.toy.core.record.key.generator.KeyGenerator;
import com.example.toy.core.utils.SchemaHelper;
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStream;
import java.nio.file.Files;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import lombok.SneakyThrows;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * ToyConfig
 *
 * @author shenbing
 * @date 2021/12/17
 */
public class ToyTableConfig extends SimpleConfig {

  private static final ConcurrentHashMap<String, ToyTableConfig> PATH_TO_CONFIG =
      new ConcurrentHashMap<>();

  /**
   * 从文件中加载配置信息
   *
   * @param file
   * @return
   */
  @SneakyThrows
  public static ToyTableConfig fromFile(File file) {
    try (BufferedReader reader = Files.newBufferedReader(file.toPath()); ) {
      Properties properties = new Properties();
      properties.load(reader);
      return new ToyTableConfig(properties);
    }
  }

  /**
   * 从路径加载配置信息
   *
   * @param path
   * @return
   */
  @SneakyThrows
  public static ToyTableConfig fromPath(Path path) {
    FileSystem fs = path.getFileSystem(new Configuration());
    try (InputStream in = fs.open(path)) {
      Properties properties = new Properties();
      properties.load(in);
      ToyTableConfig config = new ToyTableConfig(properties);
      PATH_TO_CONFIG.put(path.toString(), config);
      return config;
    }
  }

  public static ToyTableConfig fromCacheOrPath(Path path) {
    ToyTableConfig config = PATH_TO_CONFIG.get(path);
    if (config == null) {
      return fromPath(path);
    } else {
      return config;
    }
  }

  public ToyTableConfig() {
    this(new Properties());
  }

  public ToyTableConfig(Properties properties) {
    super(properties);
  }

  public String getTableName() {
    return getString(ToyConfigs.TABLE_NAME.getKey());
  }

  public boolean isArchived() {
    return getBool(ToyConfigs.ARCHIVED_ENABLED.getKey(), false);
  }

  public List<String> getPartitionFields() {
    return getTrimmedList(ToyConfigs.PARTITION_FIELDS.getKey());
  }

  public List<String> getRecordKeyFields() {
    return getTrimmedList(ToyConfigs.PRIMARY_KEY_FIELDS.getKey());
  }

  public Schema getSchema() {
    String schemaString = getString(ToyConfigs.TABLE_SCHEMA.getKey());
    return SchemaHelper.fromString(schemaString);
  }

  public Optional<Path> getDataDir() {
    Optional<String> path = getStringOptional(ToyConfigs.DATA_DIR.getKey());
    return path.map(Path::new);
  }

  @SneakyThrows
  public KeyGenerator getPartitionKeyGenerator() {
    String clzName =
        getString(
            ToyConfigs.PARTITION_KEY_GENERATOR.getKey(),
            ToyConfigs.PARTITION_KEY_GENERATOR.getDefaultValueAsString());
    return (KeyGenerator) Class.forName(clzName).newInstance();
  }

  @SneakyThrows
  public Class<PartitionPathStyle> getPartitionPathStyleClass() {
    String clzName =
        getString(
            ToyConfigs.PARTITION_PATH_STYLE_GENERATOR.getKey(),
            ToyConfigs.PARTITION_PATH_STYLE_GENERATOR.getDefaultValueAsString());
    Class<PartitionPathStyle> clz = (Class<PartitionPathStyle>) Class.forName(clzName);
    return clz;
  }
}
